package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.heima.article.service.IHotArticleService;
import com.heima.behavior.dto.ArticleOperationMessage;
import com.heima.behavior.entity.ArticleAggregatorResult;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class HotArticleResultListener {
    @Autowired
    private IHotArticleService hotArticleService;

    /**
     * 实时监听实时流式计算热点文章增量值
     *
     * @param record
     */
    @KafkaListener(topics = "${topic.hotArticleResultTopic}")
    public void handleMessage(ConsumerRecord<String, String> record){
        if (StringUtils.isNotBlank(record.value())){
            log.info("[文章服务监听到实时计算结果]：{}",record.value());
//            获取消息，将消息转换成ArticleOperationMessage
            ArticleAggregatorResult articleAggregatorResult = JSON.parseObject(record.value(), ArticleAggregatorResult.class);
//            更新Redis缓存
            hotArticleService.updateCacheAndDb(articleAggregatorResult);

        }
    }
}
